package com.nami.gate.middleware.mq.listener;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;

import com.nami.gate.api.request.message.DownMsgBatchReq;
import com.nami.gate.cache.LocalChannelCache;
import com.nami.gate.external.basic.constants.DownTopicConst;
import com.nami.gate.utils.JsonUtils;
import com.nami.im.network.model.packet.v2.Packet;
import com.nami.transport.proto.codec.MqTransMsg;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

import static com.nami.gate.external.basic.constants.CommonConst.DELTA;


public class DownMsgFlushBatchConsumer extends AbstractConsumer {

    private static final Logger log = LoggerFactory.getLogger(DownMsgFlushBatchConsumer.class);

    @Override
    public void onMessage(byte[] body) {
        try {
            TimeInterval timer = DateUtil.timer();

            MqTransMsg.BatchDispatch2GatePublishResponse response = MqTransMsg.BatchDispatch2GatePublishResponse.parseFrom(body);

            if ((System.currentTimeMillis() - response.getTimestamp()) > DELTA) {
                log.warn("[flushMore] 过期批量群推送不予消费。");
                return;
            }

            List<MqTransMsg.UserPublishMsgCO> userMsgList = response.getUserMsgList();
            log.info("[flushMore] 批量发送 size:{}", userMsgList.size());
            List<String> error = new ArrayList<>();

            if (CollectionUtil.isEmpty(userMsgList)) {
                return;
            }
            MqTransMsg.MessageHeader msgHeader = response.getMsgHeader();

            for (MqTransMsg.UserPublishMsgCO item : userMsgList) {
                try {
                    Channel channel = LocalChannelCache.get(item.getSessionId());
                    if (channel == null) {
                        log.error("[flushMore] 发送消息失败 通道丢失。 sessionId:{}", item.getSessionId());
                        continue;
                    }
                    Packet packet = new Packet();
                    packet.setCmd(msgHeader.getCmd().toByteArray()[0]);
                    packet.setQos(msgHeader.getQos().toByteArray()[0]);
                    packet.setVersion(msgHeader.getVersion().toByteArray()[0]);
                    packet.setBody(item.getPublish().toByteArray());
                    channel.writeAndFlush(packet);
                } catch (Exception e) {
                    e.printStackTrace();
                    log.error("[flushMore] 向[{}]发送消息失败:", item.getSessionId(), e.getMessage());
                    error.add(item.getSessionId());
                } // end if catch exception
            } // end for loop

            log.warn("[flushMore] 群消息发送完毕 总数={}, 失败数：{}, use:{}ms", userMsgList.size(), error.size(), timer.interval());
        } catch (Exception var3) {
            var3.printStackTrace();
            log.error("[down-login] 消费出现问题:{}", var3.getMessage());
        }

    }

    @Override
    public String getTopicName() {
        return DownTopicConst.FLUSH_MORE_TOPIC;
    }

}


